﻿
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using MX.Core.Caching.Redis.Sider.Executors;
using MX.Core.Caching.Redis.Configuration;


namespace MX.Core.Caching.Redis.Sider
{
	// REF: http://redis.io/commands
	public partial class SiderRedisClient  : IRedisClient
	{
		// TODO: IocpExecutor?
		public IEnumerable<T> Pipeline<T> (Action<IRedisClient> pipelinedCalls)
		{ 

			using (var socketIO = SockIOPool.GetInstance ().GetSock ("Pipeline")) {
				BufferedStream stream = new BufferedStream (socketIO.Stream, BufferConfig.WriteBufferSize);
				// execute in pipelined mode
				var result = SwitchExecutor<PipelinedExecutor> (new PipelinedExecutor (stream)).ExecuteRange<T> (pipelinedCalls);

				// end pipelined mode
				SwitchExecutor<ImmediateExecutor> (new ImmediateExecutor ());
				return result;
				//return null;
			} 
		}

		public byte[] Custom (string command,
		                      Action<ProtocolWriter> writeAction,
		                      Func<ProtocolReader,byte[] > readAction)
		{
			 
			var inv = Invocation<byte[]>.New (command, (p, s) => {
				writeAction (p);
			}, (r, s) => {
				return readAction (r);
			});
			return invoke (inv);
		 
		}


		#region Server

		public bool BgRewriteAof ()
		{
			return invoke ("BGREWRITEAOF",
				(r, s) => r.ReadStatus (s, "Background append only file rewriting started"));
		}

		public bool BgSave ()
		{
			return invoke ("BGSAVE", (r, s) => r.ReadStatus (s, "Background saving started"));
		}

		public KeyValuePair<string, string>[] ConfigGet (string param)
		{
			return invoke ("CONFIG", 2,
				(w, s) => {
					w.WriteArg (s, "GET");
					w.WriteArg (s, param);
				},
				(r, s) => r.ReadStrKeyValues (s));
		}

		public bool ConfigSet (string param, string value)
		{
			return invoke ("CONFIG", 3,
				(w, s) => {
					w.WriteArg (s, "SET");
					w.WriteArg (s, param);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadOk (s));
		}

		public bool ConfigResetStat ()
		{
			return invoke ("CONFIG", "RESETSTAT", (r, s) => r.ReadOk (s));
		}

		public int DbSize ()
		{
			return invoke ("DBSIZE", (r, s) => r.ReadInt (s));
		}

		public string DebugObject (string key)
		{
			return invoke ("DEBUG", 2,
				(w, s) => {
					w.WriteArg (s, "OBJECT");
					w.WriteArg (s, key);
				},
				(r, s) => r.ReadStatus (s));
		}

		public void DebugSegfault ()
		{
			invoke ("DEBUG", "SEGFAULT", (r, s) => (object)null);
		}

		public bool FlushAll ()
		{
			return invoke ("FLUSHALL", (r, s) => r.ReadOk (s));
		}

		public bool FlushDb ()
		{
			return invoke ("FLUSHDB", (r, s) => r.ReadOk (s));
		}

		public DateTime LastSave ()
		{
			return invoke ("LASTSAVE", (r, s) => r.ReadDateTime (s));
		}

		public IObservable<string> Monitor ()
		{
			using (var socketIO = SockIOPool.GetInstance ().GetSock ("Pipeline")) {
				BufferedStream stream = new BufferedStream (socketIO.Stream, BufferConfig.WriteBufferSize);
				invoke ("MONITOR", (r, s) => { 
					return r.ReadOk (stream);
				}); 
				return SwitchExecutor<MonitorExecutor> (new MonitorExecutor ()).BuildObservable (stream);
			} 

		}

		public bool Save ()
		{
			return invoke ("SAVE", (r, s) => r.ReadOk (s));
		}

		public void Shutdown ()
		{
			invoke ("SHUTDOWN");
		}

		// TODO: section support
		public KeyValuePair<string, string>[] Info ()
		{
			return invoke ("INFO", (r, s) => {
				var rawResult = r.ReadStrBulk (s)
          .Split (new[] { '\r', '\n', ':' }, StringSplitOptions.RemoveEmptyEntries);

				var result = new KeyValuePair<string, string>[rawResult.Length / 2];
				var resultIdx = 0;
				for (var i = 0; i < rawResult.Length; i += 2)
					result [resultIdx++] = new KeyValuePair<string, string> (
						rawResult [i], rawResult [i + 1]);

				return result;
			});
		}

		public bool SlaveOf (string host, int port)
		{
			return invoke ("SLAVEOF", 2,
				(w, s) => {
					w.WriteArg (s, host);
					w.WriteArg (s, port);
				},
				(r, s) => r.ReadOk (s));
		}

		#endregion

		#region Connection

		public bool Auth (string password)
		{
			return invoke ("AUTH", password, (r, s) => r.ReadOk (s));
		}

		public string Echo (string msg)
		{
			return invoke ("ECHO", msg, (r, s) => r.ReadStrBulk (s));
		}

 
		public bool Ping ()
		{
			return invoke ("PING", 0,
				(w, s) => {
				},
				(r, s) => r.ReadStatus (s, "PONG"));
		}

		public void Quit ()
		{
			invoke ("QUIT", (r, s) => r.ReadOk (s));
			Dispose ();
		}

		public bool Select (int dbIndex)
		{
			return invoke ("SELECT", 1,
				(w, s) => w.WriteArg (s, dbIndex),
				(r, s) => r.ReadOk (s));
		}

		#endregion

		#region Transactions

		public bool Multi ()
		{
			using (var socketIO = SockIOPool.GetInstance ().GetSock ("Multi")) {
				BufferedStream stream = new BufferedStream (socketIO.Stream, BufferConfig.WriteBufferSize);
				var result = invoke ("MULTI", (r, s) => r.ReadOk (s)); 
				SwitchExecutor<TransactedExecutor> (new TransactedExecutor (stream));
				return result;
			}

		}

		public bool Discard ()
		{
			var te = Executor as TransactedExecutor;
			if (te != null) {
				te.Discard ();
				SwitchExecutor<ImmediateExecutor> (new ImmediateExecutor ());
			}

			return invoke ("DISCARD", (r, s) => r.ReadOk (s));
		}

		public IEnumerable<T> Exec<T> ()
		{
 
			var te = Executor as TransactedExecutor;
			if (te == null) {
				// ReadStatus so error will be thrown, if any
				invoke ("EXEC", (r, s) => { 
					return r.ReadStatus (te.Stream);
				});
				return null;
			}
		
			// executor automatically issue EXEC
			var result = te.Finish<T> ();
		
			// restore execution mode
			SwitchExecutor<ImmediateExecutor> (new ImmediateExecutor());
			return result;
			 
		}

		public bool Watch (params string[] keys)
		{
			return invoke ("WATCH", keys.Length,
				(w, s) => Array.ForEach (keys, (key) => w.WriteArg (s, key)),
				(r, s) => r.ReadOk (s));
		}

		public bool Unwatch ()
		{
			return invoke ("UNWATCH", (r, s) => r.ReadOk (s));
		}

		#endregion

		#region Keys

		public int Del (params string[] keys)
		{
			return invoke ("DEL", keys.Length,
				(w, s) => Array.ForEach (keys, (key) => w.WriteArg (s, key)),
				(r, s) => r.ReadInt (s));
		}

		public bool Exists (string key)
		{
			return invoke ("EXISTS", key, (r, s) => r.ReadBool (s));
		}

		public bool Expire (string key, TimeSpan span)
		{
			return invoke ("EXPIRE", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, span);
				},
				(r, s) => r.ReadBool (s));
		}

		public bool ExpireAt (string key, DateTime time)
		{
			return invoke ("EXPIREAT", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, time);
				},
				(r, s) => r.ReadBool (s));
		}

		public string[] Keys (string pattern)
		{
			return invoke ("KEYS", pattern, (r, s) => r.ReadStrMultiBulk (s));
		}

		public bool Move (string key, int dbIndex)
		{
			return invoke ("MOVE", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, dbIndex);
				},
				(r, s) => r.ReadBool (s));
		}

		public bool Migrate (string host, int port, string key, int dbIndex, int timeout, bool copy, bool replace)
		{
			int argCount = 5 + (copy ? 1 : 0) + (replace ? 1 : 0);

			return invoke ("MIGRATE", 
				argCount, 
				(w, s) => {
					w.WriteArg (s, host);
					w.WriteArg (s, port);
					w.WriteArg (s, key);
					w.WriteArg (s, dbIndex);
					w.WriteArg (s, timeout);
					if (copy)
						w.WriteArg (s, "COPY");
					if (replace)
						w.WriteArg (s, "REPLACE");
				}, 
				(r, s) => r.ReadOk (s));
		}

		public bool Persist (string key)
		{
			return invoke ("PERSIST", (r, s) => r.ReadBool (s));
		}

		public string RandomKey ()
		{
			return invoke ("RANDOMKEY", (r, s) => r.ReadStrBulk (s));
		}

		public bool Rename (string oldKey, string newKey)
		{
			return invoke ("RENAME", 2,
				(w, s) => {
					w.WriteArg (s, oldKey);
					w.WriteArg (s, newKey);
				},
				(r, s) => r.ReadOk (s));
		}

		public bool RenameNX (string oldKey, string newKey)
		{
			return invoke ("RENAMENX", 2,
				(w, s) => {
					w.WriteArg (s, oldKey);
					w.WriteArg (s, newKey);
				},
				(r, s) => r.ReadBool (s));
		}

		// SORT key [BY pattern] [LIMIT offset count] [GET pattern [GET pattern ...]]
		//     [ASC|DESC] [ALPHA] [STORE destination]
		public byte[][] Sort (string key, string byPattern = null,
		                      int? limitOffset = null, int? limitCount = null,
		                      string[] getPattern = null, bool descending = false,
		                      bool alpha = false, string store = null)
		{
			var items = new LinkedList<string> ();
			items.AddLast (key);

			if (!string.IsNullOrEmpty (byPattern)) {
				items.AddLast ("BY");
				items.AddLast (byPattern);
			}

			if (limitOffset.HasValue || limitCount.HasValue) {
				items.AddLast ("LIMIT");
				items.AddLast (limitOffset.GetValueOrDefault (0).ToString ());
				items.AddLast (limitCount.GetValueOrDefault (int.MaxValue).ToString ());
			}

			if (getPattern != null) {
				foreach (var pattern in getPattern) {
					items.AddLast ("GET");
					items.AddLast (pattern);
				}
			}

			if (descending)
				items.AddLast ("DESC");
			if (alpha)
				items.AddLast ("ALPHA");

			if (!string.IsNullOrEmpty (store)) {
				items.AddLast ("STORE");
				items.AddLast (store);
			}

			return invoke ("SORT", items.Count,
				(w, s) => {
					foreach (var item in items)
						w.WriteArg (s, item);
				},
				_readObjs);
		}

		public TimeSpan TTL (string key)
		{
			return invoke ("TTL", key, (r, s) => r.ReadTimeSpan (s));
		}

		public RedisType Type (string key)
		{
			return invoke ("TYPE", key, (r, s) => RedisTypes.Parse (r.ReadStatus (s)));
		}

		#endregion

		#region Strings

		public int Append (string key, byte[] value)
		{
			return invoke<int> ("APPEND", key, value, (r, s) => r.ReadInt (s));
		}

		public long Decr (string key)
		{
			return invoke ("DECR", key, (r, s) => r.ReadInt64 (s));
		}

		public long DecrBy (string key, long value)
		{
			return invoke ("DECRBY", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadInt64 (s));
		}

		public byte[] Get (string key)
		{
			return invoke ("GET", key, _readObj);
		}

		public byte[] GetRaw (string key)
		{
			return invoke ("GET", key, (r, s) => r.ReadRawBulk (s));
		}

		public int GetTo (string key, Stream target)
		{
			return invoke ("GET", key, (r, s) => r.ReadStreamedBulk (s, target));
		}

		public int GetBit (string key, int offset)
		{
			return invoke ("GETBIT", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, offset);
				},
				(r, s) => r.ReadInt (s));
		}

		public byte[] GetRange (string key, int start, int end)
		{
			return invoke ("GETRANGE", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, start);
					w.WriteArg (s, end);
				},
				_readObj);
		}

		public byte[] GetRangeRaw (string key, int start, int end)
		{
			return invoke ("GETRANGE", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, start);
					w.WriteArg (s, end);
				},
				(r, s) => r.ReadRawBulk (s));
		}

		public int GetRangeTo (string key, int start, int end, Stream target)
		{
			return invoke ("GETRANGE", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, start);
					w.WriteArg (s, end);
				},
				(r, s) => r.ReadStreamedBulk (s, target));
		}

		public byte[] GetSet (string key, byte[] value)
		{
			return invoke ("GETSET", key, value, _readObj);
		}

		public long Incr (string key)
		{
			return invoke ("INCR", key, (r, s) => r.ReadInt64 (s));
		}

		public long IncrBy (string key, long value)
		{
			return invoke ("INCRBY", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadInt64 (s));
		}

		public byte[][] MGet (params string[] keys)
		{
			return invoke ("MGET", keys.Length,
				(w, s) => Array.ForEach (keys, (key) => w.WriteArg (s, key)),
				_readObjs);
		}

		public bool MSet (IEnumerable<KeyValuePair<string, byte[]>> mappings)
		{
			var arr = mappings.ToArray ();
			return invoke ("MSET", arr.Length * 2,
				(w, s) => Array.ForEach (arr, kv => {
					w.WriteArg (s, kv.Key);
					w.WriteArg (s, kv.Value);
				}),
				(r, s) => r.ReadOk (s));
		}

		public bool MSetNX (IEnumerable<KeyValuePair<string, byte[]>> mappings)
		{
			var arr = mappings.ToArray ();
			return invoke ("MSETNX", arr.Length * 2,
				(w, s) => Array.ForEach (arr, kv => {
					w.WriteArg (s, kv.Key);
					w.WriteArg (s, kv.Value);
				}),
				(r, s) => r.ReadBool (s));
		}

		public bool Set (string key, byte[] value)
		{
			return invoke ("SET", key, value, (r, s) => r.ReadOk (s));
		}

		public bool SetRaw (string key, byte[] raw)
		{
			return invoke ("SET", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, raw);
				},
				(r, s) => r.ReadOk (s));
		}

		public bool SetFrom (string key, Stream source, int count)
		{
			return invoke ("SET", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, source, count);
				},
				(r, s) => r.ReadOk (s));
		}

		public int SetBit (string key, int offset, int value)
		{
			return invoke ("SETBIT", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, offset);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadInt (s));
		}

		public bool SetEX (string key, TimeSpan ttl, byte[] value)
		{
			return invoke ("SETEX", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, ttl);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadOk (s));
		}

		public bool SetNX (string key, byte[] value)
		{
			return invoke ("SETNX", key, value, (r, s) => r.ReadBool (s));
		}

		public int SetRange (string key, int offset, byte[] value)
		{
			return invoke ("SETRANGE", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, offset);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadInt (s));
		}

		public int SetRangeRaw (string key, int offset, byte[] value)
		{
			return invoke ("SETRANGE", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, offset);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadInt (s));
		}

		public int SetRangeFrom (string key, int offset, Stream source, int count)
		{
			return invoke ("SETRANGE", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, offset);
					w.WriteArg (s, source, count);
				},
				(r, s) => r.ReadInt (s));
		}

		public int Strlen (string key)
		{
			return invoke ("STRLEN", key, (r, s) => r.ReadInt (s));
		}

		#endregion

		#region Hashes

		public bool HDel (string key, string field)
		{
			return invoke ("HDEL", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, field);
				},
				(r, s) => r.ReadBool (s));
		}

		[RedisVersion (2, 4, 0)]
		public int HDel (string key, params string[] fields)
		{
			return invoke ("HDEL", fields.Length + 1,
				(w, s) => {
					w.WriteArg (s, key);
					Array.ForEach (fields, (field) => w.WriteArg (s, field));
				},
				(r, s) => r.ReadInt (s));
		}

		public bool HExists (string key, string field)
		{
			return invoke ("HEXISTS", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, field);
				},
				(r, s) => r.ReadBool (s));
		}

		public byte[] HGet (string key, string field)
		{
			return invoke ("HGET", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, field);
				},
				_readObj);
		}

		public byte[] HGetRaw (string key, string field)
		{
			return invoke ("HGET", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, field);
				},
				(r, s) => r.ReadRawBulk (s));
		}

		public int HGetTo (string key, string field, Stream target)
		{
			return invoke ("HGET", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, field);
				},
				(r, s) => r.ReadStreamedBulk (s, target));
		}

		public KeyValuePair<string, byte[]>[] HGetAll (string key)
		{
			return invoke ("HGETALL", key, (r, s) => r.ReadSerializedKeyValues (s));
		}

		public long HIncrBy (string key, string field, long amount)
		{
			return invoke ("HINCRBY", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, field);
					w.WriteArg (s, amount);
				},
				(r, s) => r.ReadInt64 (s));
		}

		public string[] HKeys (string key)
		{
			return invoke ("HKEYS", key, (r, s) => r.ReadStrMultiBulk (s));
		}

		public int HLen (string key)
		{
			return invoke ("HLEN", key, (r, s) => r.ReadInt (s));
		}

		public byte[][] HMGet (string key, params string[] fields)
		{
			return invoke ("HMGET", fields.Length + 1,
				(w, s) => {
					w.WriteArg (s, key);
					Array.ForEach (fields, (field) => w.WriteArg (s, field));
				},
				_readObjs);
		}

		public bool HMSet (string key, IEnumerable<KeyValuePair<string, byte[]>> mappings)
		{
			var arr = mappings.ToArray ();
			return invoke ("HMSET", arr.Length * 2 + 1,
				(w, s) => {
					w.WriteArg (s, key);
					Array.ForEach (arr, kv => {
						w.WriteArg (s, kv.Key);
						w.WriteArg (s, kv.Value);
					});
				},
				(r, s) => r.ReadOk (s));
		}

		public bool HSet (string key, string field, byte[] value)
		{
			return invoke ("HSET", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, field);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadBool (s));
		}

		public bool HSetRaw (string key, string field, byte[] data)
		{
			return invoke ("HSET", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, field);
					w.WriteArg (s, data);
				},
				(r, s) => r.ReadBool (s));
		}

		public bool HSetFrom (string key, string field, Stream source, int count)
		{
			return invoke ("HSET", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, field);
					w.WriteArg (s, source, count);
				},
				(r, s) => r.ReadBool (s));
		}

		public bool HSetNX (string key, string field, byte[] value)
		{
			return invoke ("HSETNX", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, field);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadBool (s));
		}

		public byte[][] HVals (string key)
		{
			return invoke ("HVALS", key, _readObjs);
		}

		#endregion

		#region Lists

		public KeyValuePair<string, byte[]>? BLPop (string[] keys, int timeout)
		{
			return BLPop (timeout, keys);
		}

		// a slight variation from Redis doc since its easier to use params array
		public KeyValuePair<string, byte[]>? BLPop (int timeout, params string[] keys)
		{
			return invoke ("BLPOP", keys.Length + 1,
				(w, s) => {
					Array.ForEach (keys, (key) => w.WriteArg (s, key));
					w.WriteArg (s, timeout);
				},
				(r, s) => r.ReadSerializedKeyValue (s));
		}

		public KeyValuePair<string, byte[]>? BRPop (string[] keys, int timeout)
		{
			return BRPop (timeout, keys);
		}

		// a slight variation from Redis doc since its easier to use params array
		public KeyValuePair<string, byte[]>? BRPop (int timeout, params string[] keys)
		{
			return invoke ("BRPOP", keys.Length + 1,
				(w, s) => {
					Array.ForEach (keys, (key) => w.WriteArg (s, key));
					w.WriteArg (s, timeout);
				},
				(r, s) => r.ReadSerializedKeyValue (s));
		}

		public byte[] BRPopLPush (string src, string dest, int timeout)
		{
			return invoke ("BRPOPLPUSH", 3,
				(w, s) => {
					w.WriteArg (s, src);
					w.WriteArg (s, dest);
					w.WriteArg (s, timeout);
				},
				_readObj);
		}

		public byte[] LIndex (string key, int index)
		{
			return invoke ("LINDEX", 2,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, index);
				},
				_readObj);
		}

		public int LInsert (string key, byte[] pivot, byte[] value,
		                    bool afterPivot = false)
		{
			return invoke ("LINSERT", 4,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, afterPivot ? "AFTER" : "BEFORE");
					w.WriteArg (s, pivot);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadInt (s));
		}

		public int LLen (string key)
		{
			return invoke ("LLEN", key, (r, s) => r.ReadInt (s));
		}

		public byte[] LPop (string key)
		{
			return invoke ("LPOP", key, _readObj);
		}

		public int LPush (string key, byte[] value)
		{
			return invoke ("LPUSH", key, value, (r, s) => r.ReadInt (s));
		}

		[RedisVersion (2, 4, 0)]
		public int LPush (string key, params byte[][] values)
		{
			return invoke ("LPUSH", values.Length + 1,
				(w, s) => {
					w.WriteArg (s, key);
					foreach (var v in values)
						w.WriteArg (s, v);
				},
				(r, s) => r.ReadInt (s));
		}

		public int LPushX (string key, byte[] value)
		{
			return invoke ("LPUSHX", key, value, (r, s) => r.ReadInt (s));
		}

		public byte[][] LRange (string key, int minIncl, int maxIncl)
		{
			return invoke ("LRANGE", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, minIncl);
					w.WriteArg (s, maxIncl);
				},
				_readObjs);
		}

		public int LRem (string key, int count, byte[] value)
		{
			return invoke ("LREM", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, count);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadInt (s));
		}

		public bool LSet (string key, int index, byte[] value)
		{
			return invoke ("LSET", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, index);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadOk (s));
		}

		public bool LTrim (string key, int minIncl, int maxIncl)
		{
			return invoke ("LTRIM", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, minIncl);
					w.WriteArg (s, maxIncl);
				},
				(r, s) => r.ReadOk (s));
		}

		public byte[] RPop (string key)
		{
			return invoke ("RPOP", key, _readObj);
		}

		public byte[] RPopLPush (string srcKey, string destKey)
		{
			return invoke ("RPOPLPUSH", 2,
				(w, s) => {
					w.WriteArg (s, srcKey);
					w.WriteArg (s, destKey);
				},
				_readObj);
		}

		public int RPush (string key, byte[] value)
		{
			return invoke ("RPUSH", key, value, (r, s) => r.ReadInt (s));
		}

		[RedisVersion (2, 4, 0)]
		public int RPush (string key, params byte[][] values)
		{
			return invoke ("RPUSH", values.Length + 1,
				(w, s) => {
					w.WriteArg (s, key);
					foreach (var v in values)
						w.WriteArg (s, v);
				},
				(r, s) => r.ReadInt (s));
		}

		public int RPushX (string key, byte[] value)
		{
			return invoke ("RPUSHX", key, value, (r, s) => r.ReadInt (s));
		}

		#endregion

		#region Sets

		public bool SAdd (string key, byte[] value)
		{
			return invoke ("SADD", key, value, (r, s) => r.ReadBool (s));
		}

		[RedisVersion (2, 4, 0)]
		public int SAdd (string key, params byte[][] values)
		{
			return invoke ("SADD", values.Length + 1,
				(w, s) => {
					w.WriteArg (s, key);
					foreach (var v in values)
						w.WriteArg (s, v);
				},
				(r, s) => r.ReadInt (s));
		}

		public int SCard (string key)
		{
			return invoke ("SCARD", key, (r, s) => r.ReadInt (s));
		}

		public byte[][] SDiff (params string[] keys)
		{
			return invoke ("SDIFF", keys.Length,
				(w, s) => Array.ForEach (keys, (key) => w.WriteArg (s, key)),
				_readObjs);
		}

		public bool SDiffStore (string destKey, params string[] keys)
		{
			return invoke ("SDIFFSTORE", keys.Length + 1,
				(w, s) => {
					w.WriteArg (s, destKey);
					Array.ForEach (keys, (key) => w.WriteArg (s, key));
				},
				(r, s) => r.ReadOk (s));
		}

		public byte[][] SInter (params string[] keys)
		{
			return invoke ("SINTER", keys.Length,
				(w, s) => Array.ForEach (keys, (key) => w.WriteArg (s, key)),
				_readObjs);
		}

		public bool SInterStore (string destKey, params string[] keys)
		{
			return invoke ("SINTERSTORE", keys.Length + 1,
				(w, s) => {
					w.WriteArg (s, destKey);
					Array.ForEach (keys, (key) => w.WriteArg (s, key));
				},
				(r, s) => r.ReadOk (s));
		}

		public bool SIsMember (string key, byte[] value)
		{
			return invoke ("SISMEMBER", key, value, (r, s) => r.ReadBool (s));
		}

		public byte[][] SMembers (string key)
		{
			return invoke ("SMEMBERS", key, _readObjs);
		}

		public bool SMove (string srcKey, string destKey, byte[] value)
		{
			return invoke ("SMOVE", 3,
				(w, s) => {
					w.WriteArg (s, srcKey);
					w.WriteArg (s, destKey);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadBool (s));
		}

		public byte[] SPop (string key)
		{
			return invoke ("SPOP", key, _readObj);
		}

		public byte[] SRandMember (string key)
		{
			return invoke ("SRANDMEMBER", key, _readObj);
		}

		public bool SRem (string key, byte[] value)
		{
			return invoke ("SREM", key, value, (r, s) => r.ReadBool (s));
		}

		[RedisVersion (2, 4, 0)]
		public int SRem (string key, params byte[][] values)
		{
			return invoke ("SREM", values.Length + 1,
				(w, s) => {
					w.WriteArg (s, key);
					foreach (var v in values)
						w.WriteArg (s, v);
				},
				(r, s) => r.ReadInt (s));
		}

		public byte[][] SUnion (params string[] keys)
		{
			return invoke ("SUNION", keys.Length,
				(w, s) => Array.ForEach (keys, (key) => w.WriteArg (s, key)),
				_readObjs);
		}

		public bool SUnionStore (string destKey, params string[] keys)
		{
			return invoke ("SUNIONSTORE", keys.Length + 1,
				(w, s) => {
					w.WriteArg (s, destKey);
					Array.ForEach (keys, (key) => w.WriteArg (s, key));
				},
				(r, s) => r.ReadOk (s));
		}

		#endregion

		#region Sorted Sets

		public bool ZAdd (string key, double score, byte[] value)
		{
			return invoke ("ZADD", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, score);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadBool (s));
		}

		[RedisVersion (2, 4, 0)]
		public int ZAdd (string key, IEnumerable<KeyValuePair<double, byte[]>> values)
		{
			var arr = values.ToArray ();
			return invoke ("ZADD", arr.Length * 2 + 1, (w, s) => {
				w.WriteArg (s, key);
				foreach (var kv in arr) {
					w.WriteArg (s, kv.Key);
					w.WriteArg (s, kv.Value);
				}
			}, (r, s) => r.ReadInt (s));
		}

		public int ZCard (string key)
		{
			return invoke ("ZCARD", key, (r, s) => r.ReadInt (s));
		}

		public int ZCount (string key, double minIncl, double maxIncl)
		{
			return invoke ("ZCOUNT", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, minIncl);
					w.WriteArg (s, maxIncl);
				},
				(r, s) => r.ReadInt (s));
		}

		public double ZIncrBy (string key, double amount, byte[] value)
		{
			return invoke ("ZINCRBY", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, amount);
					w.WriteArg (s, value);
				},
				(r, s) => r.ReadDouble (s));
		}

		public int ZInterStore (string destKey, params string[] srcKeys)
		{
			return invoke ("ZINTERSTORE", srcKeys.Length + 1,
				(w, s) => {
					w.WriteArg (s, destKey);
					Array.ForEach (srcKeys, (key) => w.WriteArg (s, key));
				},
				(r, s) => r.ReadInt (s));
		}

		public int ZInterStore (string destKey, string[] srcKeys,
		                        int[] weights = null, Aggregate aggregate = Aggregate.None)
		{
			var items = new LinkedList<string> ();
			items.AddLast (destKey);
			Array.ForEach (srcKeys, k => items.AddLast (k));

			if (weights != null) {
				items.AddLast ("WEIGHTS");
				Array.ForEach (weights, w => items.AddLast (w.ToString ()));
			}

			if (aggregate != Aggregate.None) {
				items.AddLast ("AGGREGATE");
				items.AddLast (aggregate.ToString ());
			}

			return invoke ("ZINTERSTORE", items.Count,
				(w, s) => {
					foreach (var item in items)
						w.WriteArg (s, item);
				},
				(r, s) => r.ReadInt (s));
		}

		public byte[][] ZRange (string key, int startRank, int endRank)
		{
			return invoke ("ZRANGE", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, startRank);
					w.WriteArg (s, endRank);
				},
				_readObjs);
		}

		public KeyValuePair<byte[], double>[] ZRange (string key,
		                                              int startRank, int endRank, bool withScores)
		{
			// revert to normal ZRange mode when withScores = false;
			if (!withScores)
				return kvBox (ZRange (key, startRank, endRank));

			return invoke ("ZRANGE", 4,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, startRank);
					w.WriteArg (s, endRank);
					w.WriteArg (s, "WITHSCORES");
				},
				(r, s) => r.ReadSerializedWithScores (s));
		}

		public byte[][] ZRangeByScore (string key, double minIncl, double maxIncl,
		                               int? limitOffset = null, int? limitCount = null)
		{
			var includeLimit = (limitOffset.HasValue || limitCount.HasValue);

			return invoke ("ZRANGEBYSCORE", (includeLimit ? 6 : 3),
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, minIncl);
					w.WriteArg (s, maxIncl);
					if (!includeLimit)
						return;

					w.WriteArg (s, "LIMIT");
					w.WriteArg (s, limitOffset.GetValueOrDefault (0));
					w.WriteArg (s, limitCount.GetValueOrDefault (int.MaxValue));
				},
				_readObjs);
		}

		public KeyValuePair<byte[], double>[] ZRangeByScore (string key,
		                                                     double minIncl, double maxIncl,
		                                                     bool withScores,
		                                                     int? limitOffset = null, int? limitCount = null)
		{
			// revert to normal ZRangeByScore when withScores = false
			if (!withScores)
				return kvBox (ZRangeByScore (key, minIncl, maxIncl, limitOffset, limitCount));

			var includeLimit = (limitOffset.HasValue || limitCount.HasValue);

			return invoke ("ZRANGEBYSCORE", (includeLimit ? 7 : 4),
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, minIncl);
					w.WriteArg (s, maxIncl);
					w.WriteArg (s, "WITHSCORES");
					if (!includeLimit)
						return;

					w.WriteArg (s, "LIMIT");
					w.WriteArg (s, limitOffset.GetValueOrDefault (0));
					w.WriteArg (s, limitCount.GetValueOrDefault (int.MaxValue));
				},
				(r, s) => r.ReadSerializedWithScores (s));
		}

		public int ZRank (string key, byte[] value)
		{
			return invoke ("ZRANK", key, value, (r, s) => r.ReadInt (s));
		}

		public bool ZRem (string key, byte[] value)
		{
			return invoke ("ZREM", key, value, (r, s) => r.ReadBool (s));
		}

		[RedisVersion (2, 4, 0)]
		public int ZRem (string key, params byte[][] values)
		{
			return invoke ("ZREM", values.Length + 1,
				(w, s) => {
					w.WriteArg (s, key);
					foreach (var v in values)
						w.WriteArg (s, v);
				},
				(r, s) => r.ReadInt (s));
		}

		public int ZRemRangeByRank (string key, int startRank, int endRank)
		{
			return invoke ("ZREMRANGEBYRANK", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, startRank);
					w.WriteArg (s, endRank);
				},
				(r, s) => r.ReadInt (s));
		}

		public int ZRemRangeByScore (string key, double minIncl, double maxIncl)
		{
			return invoke ("ZREMRANGEBYSCORE", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, minIncl);
					w.WriteArg (s, maxIncl);
				},
				(r, s) => r.ReadInt (s));
		}

		public byte[][] ZRevRange (string key, int startRank, int endRank)
		{
			return invoke ("ZREVRANGE", 3,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, startRank);
					w.WriteArg (s, endRank);
				},
				_readObjs);
		}

		public KeyValuePair<byte[], double>[] ZRevRange (string key, int startRank, int endRank,
		                                                 bool withScores)
		{
			if (!withScores)
				return kvBox (ZRevRange (key, startRank, endRank));

			return invoke ("ZREVRANGE", 4,
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, startRank);
					w.WriteArg (s, endRank);
					w.WriteArg (s, "WITHSCORES");
				},
				(r, s) => r.ReadSerializedWithScores (s));
		}

		public byte[][] ZRevRangeByScore (string key, double minIncl, double maxIncl,
		                                  int? limitOffset = null, int? limitCount = null)
		{
			var includeLimit = (limitOffset.HasValue || limitOffset.HasValue);

			return invoke ("ZREVRANGEBYSCORE", (includeLimit ? 6 : 3),
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, minIncl);
					w.WriteArg (s, maxIncl);
					if (!includeLimit)
						return;

					w.WriteArg (s, "LIMIT");
					w.WriteArg (s, limitOffset.GetValueOrDefault (0));
					w.WriteArg (s, limitCount.GetValueOrDefault (int.MaxValue));
				},
				_readObjs);
		}

		public KeyValuePair<byte[], double>[] ZRevRangeByScore (string key,
		                                                        double minIncl, double maxIncl,
		                                                        bool withScores, int? limitOffset = null, int? limitCount = null)
		{
			if (!withScores)
				return kvBox (ZRevRangeByScore (key, minIncl, maxIncl, limitOffset, limitCount));

			var includeLimit = (limitOffset.HasValue || limitCount.HasValue);

			return invoke ("ZREVRANGEBYSCORE", (includeLimit ? 7 : 4),
				(w, s) => {
					w.WriteArg (s, key);
					w.WriteArg (s, minIncl);
					w.WriteArg (s, maxIncl);
					w.WriteArg (s, "WITHSCORES");
					if (!includeLimit)
						return;

					w.WriteArg (s, "LIMIT");
					w.WriteArg (s, limitOffset.GetValueOrDefault (0));
					w.WriteArg (s, limitCount.GetValueOrDefault (int.MaxValue));
				},
				(r, s) => r.ReadSerializedWithScores (s));
		}

		public int ZRevRank (string key, byte[] value)
		{
			return invoke ("ZREVRANK", key, value, (r, s) => r.ReadInt (s));
		}

		public double? ZScore (string key, byte[] value)
		{
			return invoke ("ZSCORE", key, value, (r, s) => r.ReadDoubleOrNull (s));
		}

		public int ZUnionStore (string destKey, params string[] srcKeys)
		{
			return invoke ("ZUNIONSTORE", srcKeys.Length + 1,
				(w, s) => {
					w.WriteArg (s, destKey);
					w.WriteArg (s, srcKeys.Length);
					Array.ForEach (srcKeys, (key) => w.WriteArg (s, key));
				},
				(r, s) => r.ReadInt (s));
		}

		public int ZUnionStore (string destKey, string[] srcKeys,
		                        int[] weights = null, Aggregate aggregate = Aggregate.None)
		{
			var items = new LinkedList<string> ();
			items.AddLast (destKey);
			items.AddLast (srcKeys.Length.ToString ());
			Array.ForEach (srcKeys, k => items.AddLast (k));

			if (weights != null) {
				items.AddLast ("WEIGHTS");
				Array.ForEach (weights, w => items.AddLast (w.ToString ()));
			}

			if (aggregate != Aggregate.None) {
				items.AddLast ("AGGREGATE");
				items.AddLast (aggregate.ToString ());
			}

			return invoke ("ZUNIONSTORE", items.Count,
				(w, s) => {
					foreach (var item in items)
						w.WriteArg (s, item);
				},
				(r, s) => r.ReadInt (s));
		}



		private KeyValuePair<byte[], double>[] kvBox (byte[][] values)
		{
			if (values == null)
				return null;

			var box = new KeyValuePair<byte[], double>[values.Length];
			for (var i = 0; i < values.Length; i++)
				box [i] = new KeyValuePair<byte[], double> (values [i], default(double));

			return box;
		}

		#endregion

		#region Pub/Sub

		public IObservable<Message> PSubscribe (params string[] keys)
		{
			return pubsubAction ("PSUBSCRIBE", keys, e => e.ActivePatterns.Add);
		}

		public int Publish (string channel, byte[] msg)
		{
			return invoke ("PUBLISH", channel, msg, (r, s) => r.ReadInt (s));
		}

		public IObservable<Message > PUnsubscribe (params string[] keys)
		{
			return pubsubAction ("PUNSUBSCRIBE", keys, e => e.ActivePatterns.Remove);
		}

		public IObservable<Message > Subscribe (params string[] keys)
		{
			return pubsubAction ("SUBSCRIBE", keys, e => e.ActiveChannels.Add);
		}

		public IObservable<Message > Unsubscribe (params string[] keys)
		{
			return pubsubAction ("UNSUBSCRIBE", keys, e => e.ActivePatterns.Remove);
		}


		private IObservable<Message > pubsubAction (string command,
		                                            string[] keys, Func<PubSubExecutor , Func<string, bool>> executorAct)
		{
			using (var socketIO = SockIOPool.GetInstance ().GetSock ("Pipeline")) {
				var pe = Executor as PubSubExecutor;
				if (pe == null)
					pe = SwitchExecutor (new PubSubExecutor (
						() => SwitchExecutor<ImmediateExecutor> (new ImmediateExecutor())));
		
				Array.ForEach (keys, k => executorAct (pe) (k));
		
		
				invoke (command, keys.Length,
					(w, s) => Array.ForEach (keys, (key) => w.WriteArg (s, key)),
					(r, s) => (object)null);
		
				return pe.GetOrBuildObservable (socketIO.Stream);
			}
		}

		#endregion


	}
}
